package com.bw.yk03;

import com.alibaba.fastjson.JSON;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

//id line
public class FlinkCDC2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DebeziumSourceFunction<String> source = MySQLSource.<String>builder()
                .hostname("hadoop-single")
                .port(3306)
                .databaseList("gmall2021") // set captured database
                .tableList("gmall2021.user_info,gmall2021.sku_info,gmall2021.base_province")
                .username("root")
                .password("root")
                .deserializer(new MyDeserializationSchemaFunction()) // converts SourceRecord to JSON String
                .build();

        DataStreamSource<String> stream = env.addSource(source);
        //stream.print();
        //分流：
        OutputTag<String> skuTag = new OutputTag<String>("sku"){};
        OutputTag<String> provTag = new OutputTag<String>("province"){};
        SingleOutputStreamOperator<String> userDS = stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                String table = JSON.parseObject(s).getString("table");
                if (table.equals("sku_info")) {
                    context.output(skuTag, s);
                } else if (table.equals("base_province")) {
                    context.output(provTag, s);
                } else {
                    collector.collect(s);
                }

            }
        });
        userDS.addSink(new FlinkCDC2SinkHbase("dim_user_info"));
        userDS.getSideOutput(skuTag).addSink(new FlinkCDC2SinkHbase("dim_sku_info"));
        userDS.getSideOutput(provTag).addSink(new FlinkCDC2SinkHbase("dim_base_province"));
        env.execute();
    }
}
